/*
 * Copyright 2013 Eduard Sedov.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.es.util;

import java.io.File;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import org.es.io.Page;
import org.es.io.PagedFile;
import static org.es.util.LinearHashing.BUCKET_SIZE;

/**
 *
 * @author Eduard Sedov
 */
public class BucketStorage {

    private final static int BUCKET_COUNT_OFFSET = 0;
    private final static int FREE_BUCKET_OFFSET = 4;
    private PagedFile mainFile;
    private PagedFile overflowFile;
    private ByteBuffer overflowHeader;
    private int cacheCapacity = 10;
    private Map<Integer, Bucket> mainCache = new HashMap();
    private Map<Integer, OverflowBucket> overflowCache = new HashMap();

    public BucketStorage(String dir) throws IOException {
        File directory = new File(dir);
        directory.mkdirs();
        File file = File.createTempFile("linearHashing", "", directory);
        mainFile = new PagedFile(file, BUCKET_SIZE);

        overflowFile = new PagedFile(new File(file.getAbsolutePath() + "_of"), BUCKET_SIZE);
        // init header
        overflowHeader = ByteBuffer.allocate(8);
        overflowHeader.putInt(BUCKET_COUNT_OFFSET, 0);
        overflowHeader.putInt(FREE_BUCKET_OFFSET, 0);
    }

    public void close() throws IOException {
        mainFile.close();
        mainFile.file().delete();
        overflowFile.close();
        overflowFile.file().delete();
    }

    public Bucket getBucket(int index, boolean reinit) {
        Bucket result = mainCache.get(index);
        if (result == null) {
            if (reinit) {
                result = Bucket.create(mainFile.get(index));
            } else {
                result = Bucket.get(mainFile.get(index));
            }
            if (mainCache.size() == cacheCapacity) {
                Iterator<Map.Entry<Integer, Bucket>> iter = mainCache.entrySet().iterator();
                if (iter.hasNext()) {
                    Map.Entry<Integer, Bucket> mapEntry = iter.next();
                    flush(mapEntry.getValue());
                    iter.remove();
                }
            }
            if (mainCache.size() < cacheCapacity) {
                mainCache.put(index, result);
            }
        } else if (reinit) {
            result = Bucket.create(result.buffer());
        }
        result.bucketIndex(index);
        return result;
    }

    public Bucket getOverflowBucket(Bucket bucket) {
        int bucketIndex = bucket.nextBucket();

        OverflowBucket result = overflowCache.get(bucketIndex);

        if (result == null) {

            if (bucketIndex > 0) {
                result = OverflowBucket.get(overflowFile.get(bucketIndex));
            } else {
                bucketIndex = overflowHeader.getInt(FREE_BUCKET_OFFSET);
                if (bucketIndex == 0) {
                    // if no free bucket exist, allocate next one and increase bucket_count
                    int count = overflowHeader.getInt(BUCKET_COUNT_OFFSET);
                    count++;
                    overflowHeader.putInt(BUCKET_COUNT_OFFSET, count);

                    bucketIndex = count;
                    result = OverflowBucket.create(overflowFile.get(bucketIndex));
                    result.bucketIndex(bucketIndex);
                } else {
                    result = OverflowBucket.get(overflowFile.get(bucketIndex));
                    overflowHeader.putInt(FREE_BUCKET_OFFSET, result.nextBucket());
                }

                bucket.nextBucket(result.bucketIndex());
                result.nextBucket(0);
                result.ownerIndex(bucket.bucketIndex() * (bucket instanceof OverflowBucket ? -1 : 1));
            }
            if (overflowCache.size() == cacheCapacity) {
                Iterator<Map.Entry<Integer, OverflowBucket>> iter = overflowCache.entrySet().iterator();
                if (iter.hasNext()) {
                    Map.Entry<Integer, OverflowBucket> mapEntry = iter.next();
                    flush(mapEntry.getValue());
                    iter.remove();
                }
            }
            if (overflowCache.size() < cacheCapacity) {
                overflowCache.put(bucketIndex, result);
            }
        }

        return result;
    }

    public void releaseBucket(Bucket bucket) {
        if (bucket.nextBucket() > 0) {
            Bucket overflowBucket = getOverflowBucket(bucket);
            releaseOverflowBucket(overflowBucket);
        }
        int bucketIndex = bucket.bucketIndex();
        if (bucketIndex == mainFile.size() - 1) {
            mainFile.truncate(bucketIndex);
        }
    }

    public void releaseOverflowBucket(Bucket bucket) {
        if (bucket instanceof OverflowBucket) {

            int ownerIndex = ((OverflowBucket) bucket).ownerIndex();

            if (ownerIndex < 0) {
                OverflowBucket ownerBucket = overflowCache.get(ownerIndex * -1);
                if (ownerBucket != null) {
                    ownerBucket.nextBucket(0);
                } else {
                    ownerBucket = OverflowBucket.get(overflowFile.get(ownerIndex * -1));
                    ownerBucket.nextBucket(0);
                    flush(ownerBucket);
                }
            } else {
                Bucket ownerBucket = mainCache.get(ownerIndex);
                if (ownerBucket != null) {
                    ownerBucket.nextBucket(0);
                } else {
                    ownerBucket = Bucket.get(mainFile.get(ownerIndex));
                    ownerBucket.nextBucket(0);
                    flush(ownerBucket);
                }
            }

            overflowHeader.putInt(FREE_BUCKET_OFFSET, bucket.bucketIndex());
            int nextBucket = bucket.nextBucket();
            while (nextBucket > 0) {
                bucket = overflowCache.get(nextBucket);
                if (bucket == null) {
                    bucket = Bucket.get(overflowFile.get(nextBucket));
                }
                nextBucket = bucket.nextBucket();
            }
            bucket.nextBucket(overflowHeader.getInt(FREE_BUCKET_OFFSET));
            flush(bucket);
        }
    }

    public void flush(Bucket bucket) {
        if (bucket instanceof OverflowBucket) {
            overflowFile.write(bucket.buffer(), bucket.bucketIndex());
        } else {
            mainFile.write(bucket.buffer(), bucket.bucketIndex());
        }
    }

    public int getCacheCapacity() {
        return cacheCapacity;
    }

    public void setCacheCapacity(int cacheCapacity) {
        this.cacheCapacity = cacheCapacity;
    }

    private static class OverflowBucket extends Bucket {

        public OverflowBucket(ByteBuffer buffer) {
            super(buffer);
        }

        public static OverflowBucket get(ByteBuffer buffer) {
            return new OverflowBucket(buffer);
        }

        public static OverflowBucket create(ByteBuffer buffer) {
            return new OverflowBucket(Page.initEmptyPage(buffer, TAIL_SIZE + 4));
        }

        public int ownerIndex() {
            return tail().getInt(TAIL_SIZE);
        }

        public void ownerIndex(int ownerIndex) {
            tail().putInt(TAIL_SIZE, ownerIndex);
        }
    }
}
